package io.confluent.kafka.consumer.processor;

import io.confluent.kafka.consumer.conf.Attributes;
import io.confluent.kafka.consumer.conf.PropertiesReader;
import io.confluent.kafka.consumer.entities.CurrentAlarmBean;
import io.confluent.kafka.consumer.entities.performance.PmModel;
import io.confluent.util.HttpUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.codehaus.jackson.map.ObjectMapper;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.*;

/**
 * Created by taotao on 2017/12/19.
 */
public class PmProcessor {
    public static void postData(ConsumerRecords<String, String> records) {
        Long count = new Long(0);
        List<CurrentAlarmBean> alarms = new ArrayList<CurrentAlarmBean>();
        if ("true".equals(PropertiesReader.DEBUG) && records.count() > 0) {
            System.out.println(records.count());
        }
        for (ConsumerRecord<String, String> record : records) {
            alarms.add(PmProcessor.parseObject(record.value()));
            count++;
            if (count % 50 == 0 || count == records.count()) {
                String jsons = PmProcessor.convertJson(alarms);
                if ("true".equals(PropertiesReader.DEBUG)) {
                    System.out.println(jsons);
                }
                if (!"".equals(jsons)) {
                    HttpUtil.doPost(Attributes.REQURL, jsons);
                    if ("true".equals(PropertiesReader.DEBUG)) {
                        System.out.println(count + "compeled...");
                    }
                }
                //插入完成后清空缓存
                alarms.clear();
            }
        }

    }
    public static String convertJson(List<CurrentAlarmBean> alarms) {
        String body = "";
        try {
            body = new ObjectMapper().writeValueAsString(alarms);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return body;
    }
    public static CurrentAlarmBean parseObject(String json) {
        CurrentAlarmBean alarm = new CurrentAlarmBean();
        ObjectMapper mapper = new ObjectMapper();
        PmModel pmModel = null;
        try {
            pmModel = mapper.readValue(json, PmModel.class);
            //保留原始告警消息
            alarm = generateObject(pmModel, json);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return alarm;
    }
    private static CurrentAlarmBean generateObject(PmModel pmModel, String json) {
        CurrentAlarmBean currentAlarmBean = new CurrentAlarmBean();
        Map<String, String> attributes = getAttributes(pmModel);
        int pmType = getPmType(attributes.get(Attributes.TYPE));
        if (pmType != 0) { //处理已定义事件，其它未知事件直接返回
            Date dateNow =  new Date();
            currentAlarmBean.setAlarmkey(attributes.get(Attributes.RESOURCE_ID) + "-" + String.valueOf(System.currentTimeMillis()) + "-" + UUID.randomUUID());
            //性能告警报文没有包含告警时间，只能用当前时间代替
            currentAlarmBean.setAlarmraisedtime(dateNow);
            currentAlarmBean.setAlarmraisedtimeGmt(dateNow);
            currentAlarmBean.setServertime(dateNow);
            currentAlarmBean.setCommenttime(dateNow);
            currentAlarmBean.setAlarmsource(attributes.get(Attributes.RESOURCE_ID));
            currentAlarmBean.setAlarmtype(pmType);
            currentAlarmBean.setCode(0);
            currentAlarmBean.setDatatype(0);
            currentAlarmBean.setDispproduct("string");
            currentAlarmBean.setId(getUUID());
            currentAlarmBean.setMoc(Attributes.MOC_PERFORMANCE);
            //先前状态由于两边告警含义定义不一致，因此先前状态也用当前状态设置，真正的先前状态和当前状态存储在Customattr
            currentAlarmBean.setOriginalperceivedseverity(getSeverity(pmModel.getSeverity()));
            currentAlarmBean.setPerceivedseverity(getSeverity(pmModel.getSeverity()));
            currentAlarmBean.setPosition1(attributes.get(Attributes.RESOURCE_ID));
            currentAlarmBean.setSystemtype(11);
            currentAlarmBean.setTimezoneoffset(0);
            currentAlarmBean.setAdditionaltext(pmModel.getReason());
            currentAlarmBean.setVisible(0);
            currentAlarmBean.setAckstate(2);
            //currentAlarmBean.setPosition2(attributes.get(Attributes.RESOURCE_NANE));
            //适配云管修改
            currentAlarmBean.setPosition2(pmModel.getAlarm_resource());

            currentAlarmBean.setAid("");
            currentAlarmBean.setAckuserid("");
            currentAlarmBean.setAcksystemid("");
            currentAlarmBean.setAcktime("");
            currentAlarmBean.setAckinfo("");
            currentAlarmBean.setSubposition1("string");
            currentAlarmBean.setSubname1("string");
            currentAlarmBean.setSubposition2("string");
            currentAlarmBean.setLink("string");
            currentAlarmBean.setNeip("string");
            currentAlarmBean.setStandardcause(0);
            currentAlarmBean.setReason(0);
            currentAlarmBean.setSpecificproblem("string");
            currentAlarmBean.setProduct1(0);
            currentAlarmBean.setProduct2(0);
            currentAlarmBean.setProduct3(0);
            currentAlarmBean.setProduct4(0);
            currentAlarmBean.setProduct5(0);
            currentAlarmBean.setProduct6(0);
            currentAlarmBean.setProduct7(0);
            currentAlarmBean.setProduct8(0);
            currentAlarmBean.setProduct9(0);
            currentAlarmBean.setTimezoneid("string");
            currentAlarmBean.setDstsaving(0);
            currentAlarmBean.setAlarmchangedtime("");
            currentAlarmBean.setCleartype(0);
            currentAlarmBean.setClearuserid("string");
            currentAlarmBean.setClearsystemid("string");
            currentAlarmBean.setAlarmclearedtime("");
            currentAlarmBean.setAlarmclearedtimeGmt("");
            currentAlarmBean.setCommenttext("string");
            currentAlarmBean.setCommentuserid("string");
            currentAlarmBean.setCommentsystemid("string");
            //currentAlarmBean.setCustomattr1("string");
            currentAlarmBean.setCustomattr1(pmModel.getAlarm_meter());
            currentAlarmBean.setCustomattr2(pmModel.getAlarm_description());
            //存储先前状态，OK，Alarm或者insufficient data
            currentAlarmBean.setCustomattr3(pmModel.getPrevious());
            //存储当前状态，OK，Alarm或者insufficient data
            currentAlarmBean.setCustomattr4(pmModel.getCurrent());
            //存储alarm_id
            currentAlarmBean.setCustomattr5(pmModel.getAlarmId());
            //存储type
            currentAlarmBean.setCustomattr6(pmModel.getReasonData().getType());
            //存储most_recent
            currentAlarmBean.setCustomattr7(pmModel.getReasonData().getMostRecent());
            //存储disposition
            currentAlarmBean.setCustomattr8(pmModel.getReasonData().getDisposition());
            currentAlarmBean.setCustomattr9("string");
            currentAlarmBean.setCustomattr10("string");
            currentAlarmBean.setCustomattr11("string");
            currentAlarmBean.setCustomattr12("string");
            currentAlarmBean.setAdmc(0);

            currentAlarmBean.setReserve1("string");
            currentAlarmBean.setReserve2("string");
            currentAlarmBean.setReserve3("string");
            currentAlarmBean.setReserve4("string");
            currentAlarmBean.setReserve5("string");
            //存储原始告警消息
            currentAlarmBean.setChangetrack(json);
            currentAlarmBean.setOmmid("string");
            currentAlarmBean.setPathid("string");
            currentAlarmBean.setPathname("string");
            currentAlarmBean.setAlarmreportcount(pmModel.getReasonData().getCount());
            currentAlarmBean.setPlmn1(0);
            currentAlarmBean.setPlmn2(0);
            currentAlarmBean.setPlmn3(0);
            currentAlarmBean.setPlmn4(0);
            currentAlarmBean.setPlmn5(0);
            currentAlarmBean.setPlmn6(0);
            currentAlarmBean.setIspublic2plmn(0);
            currentAlarmBean.setEmsid("string");
            currentAlarmBean.setRelationflag(0);
            currentAlarmBean.setIsnaffilter(0);
            currentAlarmBean.setThresholdinfo("string");
            currentAlarmBean.setIntermittencecount(0);
            currentAlarmBean.setDebugstate(0);
            currentAlarmBean.setSortinfo("string");
            currentAlarmBean.setClearinfo("string");
        }
        return currentAlarmBean;
    }

    private static Map<String, String> getAttributes(PmModel pmModel) {
        Map<String, String> attributes = new HashMap<String, String>();
        attributes.put(Attributes.TYPE, pmModel.getReasonData().getType());
        //Resource_id从设置告警规则时填入的值来取，格式为：cpu alarm location:vmName:940087d5-f889-4533-a5be-9574af5dd880"
        String attrs[] = pmModel.getAlarmName().split(":");
        if (attrs.length > 2) {
            attributes.put(Attributes.RESOURCE_NANE, attrs[1]);
            attributes.put(Attributes.RESOURCE_ID, attrs[2]);
        } else { //设置默认值0000
            attributes.put(Attributes.RESOURCE_NANE, "0000");
            attributes.put(Attributes.RESOURCE_ID, "0000");
        }


        return attributes;
    }

    /**
     * 判断告警类型。注意:Aodh Event针对定义的每个Openstack操作都有两条事件，.start事件和.end事件。
     * 程序只处理.end结尾的事件，.start事件不处理，因为.start事件中不包括必要的字段域，例如resource_id等。
     * @param pmType
     * @return
     */
    private static int getPmType(String pmType) {
        int code = 0;
        if (Attributes.THRESHOLD.equals(pmType)) code = 41;
        else if (Attributes.COMBINATION.equals(pmType)) code = 42;
        else  code = 0;
        return code;
    }

    private static long getUUID()
    {
        Random r = new Random();
        //如生成的随机位数不足6位则自动加零补充
        DecimalFormat g = new DecimalFormat("10000");
        //返回时间增量+随机数的序列
        return Long.parseLong(String.format("%s%s", System.currentTimeMillis(), g.format(r.nextInt(10000))));
    }

    private static int getSeverity(String severity) {
        int severityCode = 0;
        if (Attributes.SEVERITY_LOW.equals(severity)) severityCode = 3;
        else if (Attributes.SEVERITY_MODERATE.equals(severity)) severityCode = 2;
        else if (Attributes.SEVERITY_CRITICAL.equals(severity)) severityCode = 1;
        //其它类型设成不确定告警
        else severityCode = 5;
        return severityCode;
    }
}
